package com.atguigu.wc;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.operators.*;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.util.Collector;

/**
 * @author yhm
 * @create 2024-03-30 9:56
 */
public class BatchWordCount1 {
    public static void main(String[] args) throws Exception {
        // TODO1  构建Flink环境  可以区分创建批处理环境和流处理环境
        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

        // TODO2  读取数据源
        DataSource<String> lineDs = env.readTextFile("input/words.txt");

        // TODO3 处理计算
        // 3.1 切分单词
        // 可以使用lambda表达式写法  不过flink容易类型判断错误导致报错   需要在多个位置显示声明数据的类型  主要是输出数据类型
        FlatMapOperator<String, Tuple2<String, Long>> tuple2Op = lineDs.flatMap((FlatMapFunction<String, Tuple2<String, Long>>) (String value,Collector<Tuple2<String, Long>> out) -> {
            // 单词切分
            String[] words = value.split(" ");
            for (String word : words) {
                // 使用out来完成数据输出
                out.collect(new Tuple2<>(word, 1L));
            }
        }).returns(Types.TUPLE(Types.STRING,Types.LONG));

        // 3.2 聚合相同单词
        // 相对简单的数据类型  flink可以直接推断  可以写成特别简单的lambda表达式
//        UnsortedGrouping<Tuple2<String, Long>> groupByOp = tuple2Op.groupBy( value -> value.f0);

        // reduce第一个参数是上一次聚合的结果
        // 第二个参数是这一次新加的元素
        // 在第一个reduce计算中  value1就是第一个元素  value2是第二个元素
//        ReduceOperator<Tuple2<String, Long>> resultData = groupByOp.reduce((ReduceFunction<Tuple2<String, Long>>) (value1, value2) -> new Tuple2<>(value1.f0, value1.f1 + value2.f1));


        AggregateOperator<Tuple2<String, Long>> resultData = tuple2Op.groupBy(0)
                .sum(1);

        resultData.print();
    }
}
